package biz.ws.ws.sender;

import biz.ws.WsConfig;
import biz.ws.ws.*;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.kafka.core.KafkaProducerException;
import org.springframework.kafka.core.KafkaSendCallback;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.lang.NonNull;

import javax.annotation.Resource;
import java.util.Objects;

/**
 * @创建人 wsdc
 * @时间 2021/4/2
 * @描述 默认的发送方式 直接找
 */
public class KafkaSharedMsgSender implements MsgSender, InitializingBean {
    @Resource
    UserContainer userContainer;

    @Resource
    KafkaTemplate kafka;

    @Resource
    WsConfig config;

    @Override
    public void send(MsgHolder context) {
        Message msg = context.getMessage();

        MsgSession session = userContainer.get(msg.getToId());
        if(session != null){
            for (Channel value : session.getInnerMap().values()) {
                WsUtils.write(value,msg);
            }

            return ;
        }

        //  消息推送到kafka中
        Long toId = msg.getToId();
        int hash = (int) (toId % config.getSharedSize());

        if(hash == config.getShared()){
            //  是我当前这个节点的数据 但是用户没有链接
            //  或者是我之前宕机了 刚刚起来，用户还在临时节点上面
            //  所以选择不发数据
            return ;
        }

        //  分发到kafka中
        kafka.send(WsCst.TopicName.WS_SEND,hash, JSON.toJSONString(msg));
    }

    @Override
    public void send(MsgHolder context, @NonNull SendListener l) {
        Message msg = context.getMessage();

        //  如果可以取到session 那么说明这个节点是 主节点或者容灾节点
        MsgSession session = userContainer.get(msg.getToId());
        if(session != null){
            //  before 钩子
            l.beforePost();
            for (Channel value : session.getInnerMap().values()) {
                ChannelFuture f = WsUtils.write(value, msg);
                if(f == null){
                    //  失败钩子
                    l.failure();
                    return ;
                }

                //  发送消息之后的回调
                f.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if(future.isSuccess()){
                            l.success();
                        }else{
                            l.failure();
                        }
                    }
                });
            }

            return ;
        }

        //  消息推送到kafka中
        Long toId = msg.getToId();
        int hash = (int) (toId % config.getSharedSize());

        if(hash == config.getShared()){
            //  如果用户没有登录 或者用户登录到了容灾节点  此时不管  消息保存
            //  触发before 钩子
            l.beforePost();
            return ;
        }

        if(!Objects.equals(context.getSource(),WsCst.MsgSource.CLIENT)){
            //  如果不是来自客户端的消息 就不要分发kafka了
            return ;
        }

        //  分发到kafka中
        kafka.send(WsCst.TopicName.WS_SEND,hash, JSON.toJSONString(msg))
                .addCallback(new KafkaSendCallback() {
                    @Override
                    public void onFailure(KafkaProducerException ex) {
                        // TODO: 2021/4/4 失败如何处理 处理
                        ex.printStackTrace();
                    }

                    @Override
                    public void onSuccess(Object result) {
                        //  推送的kafka中成功
                    }
                });
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        int k = 0;

    }
}
